package com.example.sparkmanager.service.impl;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.example.sparkmanager.model.ExecutorJarEntity;
import com.example.sparkmanager.model.JobEntity;
import com.example.sparkmanager.model.JobQueue;
import com.example.sparkmanager.repository.ExecutorJarRepository;
import com.example.sparkmanager.repository.JobRepository;
import com.example.sparkmanager.service.JobService;
import com.example.sparkmanager.util.HttpClient;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


@Service
@RequiredArgsConstructor
@Slf4j
public class JobServiceImpl implements JobService {

    private static Boolean queueExecuteStatus = Boolean.valueOf(false);
    ExecutorService executor = Executors.newFixedThreadPool(1);
    @Value("${spark.config.server}")
    private String sparkServer;//spark服务ip地址
    @Value("${spark.config.port}")
    private String sparkPort;//spark web ui端口
    @Value("${spark.config.restfulPort}")
    private String sparkRestfulPort;//spark restful端口地址
    @Value("${spark.config.submitServer}")
    private String sparkSubmitServer;//spark://spark服务ip地址：web ui端口
    @Value("${spark.config.jarDir}")
    private String jarDir;//应用程序所在目录
    @Value("${spark.config.version}")
    private String sparkVersion;
    @Value("${spark.config.cluster.cores}")
    private Integer cores;
    @Autowired
    private HttpClient httpClient;
    @Autowired
    private JobRepository jobRepository;
    @Autowired
    private ExecutorJarRepository executorJarRepository;

    @Override
    public synchronized void jobExecute(JobEntity job) {
        try {
            log.info("@JobExecute: 开始执行任务，jobId为{}", job.getId());
            Boolean resourceEnough = checkSparkServerResource(Integer.valueOf(1), Long.valueOf(1L));
            if (!resourceEnough.booleanValue()) {
                log.info("@JobExecute: 当前Spark资源已满，将任务{}添加至等待队列中", job.getId());
                JobQueue.addJob(job);
                if (!queueExecuteStatus.booleanValue()) {
                    queueExecute();
                    queueExecuteStatus = Boolean.valueOf(true);
                }
            } else {
                log.info("@JobExecute: 开始创建Spark任务{}", job.getId());
                jobSubmit2Spark(job);
                log.info("@JobExecute: Spark任务{}提交成功", job.getId());
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("@JobExecute: error is {}", e.getMessage());
            job.setStatus("FAILED");
            job.setMessage(e.getMessage());
            this.jobRepository.save(job);
        }
    }


    /* private synchronized void queueExecute() {
         log.info("@QueueExecute: 开启Job队列监听方法");
         this.executor.execute((Runnable) new Object(this));
         log.info("@QueueExecute: 当前队列中任务已执行完毕");
     }
 */
    private synchronized void queueExecute() {
        log.info("@QueueExecute: 开启Job队列监听方法");

        this.executor.execute(() -> {
            try {
                while (true) {
                    JobEntity nextJob = JobQueue.getTopJob();
                    if (nextJob == null) {
                        log.info("@QueueExecute: 队列为空，停止监听");
                        break;
                    }

                    Boolean resourceEnough = checkSparkServerResource(1, 1L);
                    if (resourceEnough) {
                        log.info("@QueueExecute: 资源充足，执行任务 {}", nextJob.getId());
                        jobSubmit2Spark(nextJob);
                    } else {
                        log.info("@QueueExecute: 资源不足，将任务 {} 重新放回队列", nextJob.getId());
                        JobQueue.addJob(nextJob);
                        // 等待一段时间后重试
                        Thread.sleep(10000);
                    }

                    // 添加小延迟，避免CPU过度占用
                    Thread.sleep(1000);
                }
            } catch (InterruptedException e) {
                log.info("@QueueExecute: 队列监听被中断");
                Thread.currentThread().interrupt();
            } catch (Exception e) {
                log.error("@QueueExecute: 队列执行出错 - {}", e.getMessage(), e);
            } finally {
                queueExecuteStatus = false;
                log.info("@QueueExecute: 队列监听结束");
            }
        });

        log.info("@QueueExecute: 队列监听任务已启动");
    }


    //./spark-submit --class  com.mayy.spark.demo.SimpleApp  --master spark://192.40.10.130:7077  /root/myspark/jars/spark-simple-demo-0.0.1.jar  /root/myspark/jars/README.md
    private synchronized void jobSubmit2Spark(JobEntity job) throws Exception {
//        String executorId = "";
//        ExecutorJarEntity executorJar = this.executorJarRepository.findById(executorId).get();
        ExecutorJarEntity executorJar = new ExecutorJarEntity();
        executorJar.setFileName("spark-simple-demo-0.0.1.jar");
        String jarFileName = executorJar.getFileName();
        String jarPath = this.jarDir + "/" + jarFileName;

        JSONObject sparkSubmitData = new JSONObject();

        sparkSubmitData.put("action", "CreateSubmissionRequest");
        sparkSubmitData.put("appResource", jarPath);
        sparkSubmitData.put("clientSparkVersion", this.sparkVersion);
        JSONObject environmentVariables = new JSONObject();
        environmentVariables.put("SPARK_ENV_LOADED", "1");
        sparkSubmitData.put("environmentVariables", environmentVariables);
        sparkSubmitData.put("mainClass", "com.mayy.spark.demo.SimpleApp");//注意 这个入口地址

        JSONObject sparkProperties = new JSONObject();
        sparkProperties.put("spark.driver.supervise", "false");
        sparkProperties.put("spark.app.name", job.getId());
        sparkProperties.put("spark.submit.deployMode", "cluster");//表示 Spark 应用程序的部署模式为 集群模式（cluster mode）
        // spark://192.40.10.130:8080- Spark Web UI 端口
//        sparkProperties.put("spark.master", this.sparkSubmitServer);
        //spark://192.40.10.130:7077- Standalone 集群管理器端口
        sparkProperties.put("spark.master", "spark://192.40.10.130:7077");
        //--master spark://192.40.10.130:7077
        sparkProperties.put("spark.executor.memory", "512m");
        sparkProperties.put("spark.executor.cores", "1");
        sparkProperties.put("spark.cores.max", "1");
        sparkProperties.put("spark.driver.memory", "512m");
        sparkProperties.put("spark.driver.cores", "1");
        sparkProperties.put("spark.speculation", "true");
        sparkProperties.put("spark.jars", jarPath);
        sparkSubmitData.put("sparkProperties", sparkProperties);

        JSONArray appArgs = new JSONArray();
//        appArgs.add(job.getId());
        // 从 JobEntity 中获取业务参数（更灵活的方式）
        String businessParam = job.getParam(); // 假设 JobEntity 中有这个字段
        if (businessParam == null || businessParam.trim().isEmpty()) {
            businessParam = "/root/myspark/jars/README.md"; // 默认值
        }
        appArgs.add(businessParam); // 第二个参数：业务参数
        sparkSubmitData.put("appArgs", appArgs);

        Map<String, String> header = new HashMap<>();
        header.put("Content-Type", "application/json;charset=UTF-8");

        System.out.println(sparkSubmitData);

        String resultStr = this.httpClient.post("v1/submissions/create", sparkSubmitData, this.sparkServer, this.sparkRestfulPort, header,
                Long.valueOf(3000L));
        JSONObject result = JSON.parseObject(resultStr);

        Boolean status = result.getBoolean("success");
        if (!status.booleanValue()) {
            throw new Exception(result.getString("message"));
        }
        job.setStatus("RUNNING");
        this.jobRepository.save(job);
    }


    public synchronized Boolean checkSparkServerResource(Integer cores, Long memoryGB) throws Exception {
        String resultStr = this.httpClient.get("json", this.sparkServer, this.sparkPort, null);
        JSONObject sparkResource = JSON.parseObject(resultStr);
        Integer totalCores = sparkResource.getInteger("cores");
        Integer userdCores = sparkResource.getInteger("coresused");

        Long totalMemory = sparkResource.getLong("memory");
        Long usedMemory = sparkResource.getLong("memoryused");

        Integer freeCores = Integer.valueOf(totalCores.intValue() - userdCores.intValue() - cores.intValue() / 2);
        Long freeMemory = Long.valueOf(totalMemory.longValue() - usedMemory.longValue() - (cores.intValue() / 2));

        Boolean enough = Boolean.valueOf(true);
        if (cores.intValue() > freeCores.intValue() || memoryGB.longValue() * 1024L > freeMemory.longValue()) {
            enough = Boolean.valueOf(false);
        }
        return enough;
    }

}
